//package com.zn.socketReceiver;
//
//import com.alibaba.fastjson.JSONObject;
//import com.zn.entity.CThostFtdcDepthMarketDataField;
//import com.zn.entity.KLineEntity;
//import com.zn.service.MdInfoService;
//import com.zn.util.*;
//import com.zn.vo.SocketJsonData;
//import com.zn.websocket.group.ChannelGroups;
//import io.netty.buffer.ByteBuf;
//import io.netty.buffer.Unpooled;
//import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
//import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.data.mongodb.core.MongoTemplate;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.scheduling.annotation.Async;
//
//import java.io.*;
//import java.net.InetSocketAddress;
//import java.net.Socket;
//import java.nio.ByteBuffer;
//import java.nio.channels.ClosedByInterruptException;
//import java.text.SimpleDateFormat;
//import java.util.Date;
//import java.util.zip.GZIPOutputStream;
//
//public class TickDataReceiver extends DataReceiver implements Runnable {
//	private static final Logger LOGGER = LoggerFactory.getLogger(TickDataReceiver.class);
//	private static final Integer FIDLE_SIZE=30;
//	private RedisTemplate<String,String> redisTemplate;
//	private MongoTemplate mongoTemplate;
//	private MdInfoService mdInfoService;
//	protected String host;
//	protected int port;
//	protected long retryDelay;
//	protected Thread thread;
//
//	public TickDataReceiver(RedisTemplate<String, String> redisTemplate,MongoTemplate mongoTemplate,MdInfoService mdInfoService,String host,int port) {
//		this.redisTemplate = redisTemplate;
//		this.mongoTemplate=mongoTemplate;
//		this.mdInfoService=mdInfoService;
//		this.host=host;
//		this.port=port;
//	}
//
//	@Override
//	public void start() throws Exception {
//		if(this.thread != null) {
//			LOGGER.info(this.getClass() + "Proxy data receiver already started");
//			return;
//		}
////		this.host = "101.132.174.85";
////		this.port = 4590;
//		this.retryDelay = 5000;
//		LOGGER.info(this.getClass() + "Proxy data receiver start host:" + this.host + ", port:" + this.port + ", retry delay:" + this.retryDelay);
//		this.thread = new Thread(this, "TickDataReceiver");
//		this.thread.start();
//	}
//
//	@Override
//	public void stop() throws Exception {
//		if(this.thread != null) {
//			this.thread.interrupt();
//			this.thread = null;
//		}
//	}
//
//	@Override
//	public void run() {
//		LOGGER.info("执行run方法：" + Thread.interrupted());
//		while(Thread.interrupted() == false) {
//			try {
//				this.process();
//			}catch(InterruptedException ie) {
//				break;
//			}catch(ClosedByInterruptException cbie) {
//				break;
//			}catch(Exception e) {
//				LOGGER.info(this.getClass() + "" + e);
//			}
//			try {
//				Thread.sleep(this.retryDelay);
//				LOGGER.info(this.getClass() + "Proxy data receiver retry");
//			}catch (InterruptedException e) {
//				break;
//			}
//		}
//		LOGGER.info(this.getClass() + "Proxy data receiver end");
//	}
//
//	public void process() throws Exception {
//		// Prepare date format
//		SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd HH:mm:ss.SSS");
//		dateFormat.setTimeZone(Constant.DEFAULT_TIMEZONE);
//		// Prepare read buffer
//		ByteBuffer bufReceive = ByteBuffer.allocate(1024 * 200);
//		byte bufRead[] = new byte[1024 * 100];
//		// Prepare socket
//		Socket socket = new Socket();
//		socket.setSoTimeout(60000);
//		socket.connect(new InetSocketAddress(this.host, this.port), 30000);
//		InputStream is = socket.getInputStream();
//		try {
//			int count = 0;
//			while((count = is.read(bufRead)) > 0) {
//				if(count > bufReceive.remaining()) {
//					throw new Exception("Proxy data receiver buffer overflow");
//				}
//				bufReceive.put(bufRead, 0, count);
//				this.parseData(bufReceive, dateFormat);
//			}
//		}
//		finally {
//			try {is.close();}catch(Exception e) {e.printStackTrace();}
//			try {socket.close();}catch(Exception e) {e.printStackTrace();}
//		}
//	}
//
//	/**
//	 * 解析行情数据
//	 * @param buf 数据缓冲
//	 */
//	protected void parseData(ByteBuffer buf, SimpleDateFormat dateFormat) {
//		buf.flip();
//		byte data[] = new byte[buf.remaining()];
//		buf.get(data);
//		int startIndex = 0;
//		for(int i = 0; i < data.length; i++) {
//			if(data[i] == '\n') {
//				if(i > startIndex) {
//					String item = null;
//					try {
//						item = new String(data, startIndex, i - startIndex, "GB2312");
//						if(item.startsWith("heartbeat")) {//接受心跳数据
//
//						}else if(item.startsWith("mainContract")) {//接受主力合约列表的数据
////							LOGGER.info("mainContract"+item);
//							this.parseMainContract(item);
//						}else if(item.startsWith("tick")){//接受深度行情数据
////							LOGGER.info("tick" + item);
//							this.parseDataItem(item);
//						}else if(item.startsWith("underwayKline")){//接受正在进行中的K线高开低收的值
////							LOGGER.info("underwayKline"+item);
//							this.parseUnderwayKline(item);
//						}else if(item.startsWith("kLineItem")){
////							LOGGER.info("kLineItem"+item);
//							this.parsekLineItem(item);
//						}
//					}catch(Exception e) {
//						e.printStackTrace();
//						LOGGER.info(this.getClass() + "Proxy data receiver parse error: " + e.getMessage());
//						LOGGER.info(this.getClass() + "" + item);
//					}
//				}
//				startIndex = i + 1;
//			}
//		}
//		buf.clear();
//		if(startIndex < data.length) {
//			buf.put(data, startIndex, data.length - startIndex);
//		}
//	}
//
//	/**
//	 * 处理主力合约
//	 * @param item
//	 * @throws Exception
//	 */
//	protected void parseMainContract(String item) throws Exception {
//		//mainContract|0|instrumentID|lastPrice|preSettlementPrice|openInterest
//		LOGGER.info(DateUtils.parseToDateTimeStr(new Date())+"接收到c++的主力合约列表数据--------item:"+item);
//		String [] detailArray=item.split("\\|");
//		Integer flag=Integer.valueOf(detailArray[23]);
//		if(flag==0){
//			LOGGER.info("删除Key值................");
//			//先把主力合约的key删掉
//			redisTemplate.delete(Constant.MAIN_CONTRACT_KEY);
//		}
//		//将合约code转换为合约名称
//		String goodsCode=detailArray[2].replaceAll("[^a-z^A-Z]", "").toUpperCase();
//		String instrument=detailArray[2].substring(goodsCode.length());
//		String instrumentName= CodeConvertName.convert(goodsCode)+instrument;
//		//主力合约列表
//		CThostFtdcDepthMarketDataField mainContract=new CThostFtdcDepthMarketDataField(
//				detailArray[1],detailArray[2],detailArray[3],detailArray[4],
//				detailArray[5],detailArray[6],detailArray[7],detailArray[8],
//				detailArray[9],detailArray[10],detailArray[11],detailArray[12],
//				detailArray[13],detailArray[14],detailArray[15],detailArray[16],
//				detailArray[17],detailArray[18],detailArray[19],
//				detailArray[20],detailArray[21],instrumentName,detailArray[22]
//		);
//		//主力合约json
//		Object mainContractToJson=JSONObject.toJSON(mainContract);
//		/**推送完成之后更新首页主力合约列表的合约最新值**/
//		redisTemplate.opsForHash().put(Constant.MAIN_CONTRACT_KEY, mainContract.getInstrumentID(), mainContractToJson.toString());
//		/**将主力合约推到交易服务判断最新的主力合约是否存在**/
//		redisTemplate.convertAndSend(Constant.MAIN_CONTRACT_CHANNEL, mainContract.getInstrumentID());
//	}
//
//	/**
//	 * 处理事实K线最高最低
//	 * @param item
//	 * @throws Exception
//	 */
//	protected void parsekLineItem(String item) throws Exception {
//		//kLineItem|K线类型|code|curTime(yyyy-MM-dd HH:mm:ss)|open|close|high|low|volume|tradeAmt|positionNum|productCode|productMonth
//		LOGGER.info("parsekLineItem--------item:" + item);
//		String [] detailArray=item.split("\\|");
//		Integer kLineType=Integer.valueOf(detailArray[1]);
//		//将商品CODE转为大写
//		String productCode=detailArray[11].toUpperCase();
//		String collectionName="";
//		switch (kLineType){
//			/**处理一分钟**/
//			case 1:collectionName=Constant.QUOTATIONS_KLINE_ONE_MIN+productCode;break;
//			/**处理五分钟**/
//			case 2:collectionName=Constant.QUOTATIONS_KLINE_FIVE_MIN+productCode;break;
//			/**处理十五分钟**/
//			case 3:collectionName=Constant.QUOTATIONS_KLINE_FIFTEEN_MIN+productCode; break;
//			/**处理三十分钟**/
//			case 4:collectionName=Constant.QUOTATIONS_KLINE_THIRTY_MIN; break;
//			/**处理一小时**/
//			case 5:collectionName=Constant.QUOTATIONS_KLINE_ONE_HR; break;
//			/**处理四小时**/
//			case 6:collectionName=Constant.QUOTATIONS_KLINE_FOUR_HR; break;
//			/**日线**/
//			case 7:collectionName=Constant.QUOTATIONS_KLINE_DAY; break;
//			default:
//				LOGGER.info("parsekLineItem...解析C++数据包出异常了：" + item);
//				break;
//		}
//		this.insertMongoKlineData(item, collectionName);
//	}
//
//	/**
//	 * 处理事实K线最高最低
//	 * @param item
//	 * @throws Exception
//	 */
//	protected void parseUnderwayKline(String item) throws Exception {
//		//underwayKline|k线类型|FG809|2018-08-13 11:04:13|1466.000|1466.000|1467.000|1466.000|12724|0.00|53282.000|FG|09|
//		LOGGER.info("parseUnderwayKline--------item:" + item);
//		String [] detailArray=item.split("\\|");
//		Integer kLineType=Integer.valueOf(detailArray[1]);
//		String contractCode=detailArray[2];
//		switch (kLineType){
//			/**处理一分钟**/
//			case 1:UnderwayKlineMap.UNDERWAY_KLINE_ONE_MIN.put(contractCode, item);
////				LOGGER.info("存储map后的UNDERWAY_KLINE_ONE_MIN大小："+UnderwayKlineMap.UNDERWAY_KLINE_ONE_MIN.size());
//				break;
//			/**处理五分钟**/
//			case 2:UnderwayKlineMap.UNDERWAY_KLINE_FIVE_MIN.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_FIVE_MIN："+UnderwayKlineMap.UNDERWAY_KLINE_FIVE_MIN.size());
//				break;
//			/**处理十五分钟**/
//			case 3:UnderwayKlineMap.UNDERWAY_KLINE_FIFTEEN_MIN.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_FIFTEEN_MIN："+UnderwayKlineMap.UNDERWAY_KLINE_FIFTEEN_MIN.size());
//				break;
//			/**处理三十分钟**/
//			case 4:UnderwayKlineMap.UNDERWAY_KLINE_THIRTY_MIN.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_THIRTY_MIN："+UnderwayKlineMap.UNDERWAY_KLINE_THIRTY_MIN.size());
//				break;
//			/**处理一小时**/
//			case 5:UnderwayKlineMap.UNDERWAY_KLINE_ONE_HR.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_ONE_HR：" + UnderwayKlineMap.UNDERWAY_KLINE_ONE_HR.size());
//				break;
//			/**处理四小时**/
//			case 6:UnderwayKlineMap.UNDERWAY_KLINE_FOUR_HR.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_FOUR_HR：" + UnderwayKlineMap.UNDERWAY_KLINE_FOUR_HR.size());
//				break;
//			/**日线**/
//			case 7:UnderwayKlineMap.UNDERWAY_KLINE_DAY.put(contractCode, item);
////				LOGGER.info("UNDERWAY_KLINE_DAY：" + UnderwayKlineMap.UNDERWAY_KLINE_DAY.size());
//				break;
//			default:
//				LOGGER.info("parseUnderwayKline....解析C++数据包出异常了：" + item);
//				break;
//		}
//	}
//
//	/**
//	 * 处理深度行情
//	 * @param fromCLanguageMdInfo
//	 * @throws Exception
//	 */
//	protected void parseDataItem(String fromCLanguageMdInfo) throws Exception {
////		LOGGER.info("接受到C++的行情数据："+fromCLanguageMdInfo);
//		/**根据管道符分割**/
//		String [] mdDetailArray=fromCLanguageMdInfo.split("\\|");
//		/**截取的字段符合要求才推送**/
//		if(mdDetailArray.length==FIDLE_SIZE){
//			/**String tradingDay, String instrumentID, Double lastPrice, Double preSettlementPrice, Double preClosePrice,
//			 * Double preOpenInterest, Double openPrice, Double highestPrice, Double lowestPrice, Integer volume, Double turnover,
//			 * Double openInterest, Double upperLimitPrice, Double lowerLimitPrice, String updateTime, Double bidPrice1,
//			 * Integer bidVolume1, Double askPrice1, Integer askVolume1, Double averagePrice, String actionDay**/
//			//将合约code转换为合约名称
//			String goodsCode=mdDetailArray[2].replaceAll("[^a-z^A-Z]", "").toUpperCase();
//			String instrument=mdDetailArray[2].substring(goodsCode.length());
//			String instrumentName= CodeConvertName.convert(goodsCode)+instrument;
//			CThostFtdcDepthMarketDataField marketDataField=new CThostFtdcDepthMarketDataField(
//					mdDetailArray[1],mdDetailArray[2],mdDetailArray[3],mdDetailArray[4],
//					mdDetailArray[5],mdDetailArray[6],mdDetailArray[7],mdDetailArray[8],
//					mdDetailArray[9],mdDetailArray[10],mdDetailArray[11],mdDetailArray[12],
//					mdDetailArray[13],mdDetailArray[14],mdDetailArray[15],mdDetailArray[16],
//					mdDetailArray[17],mdDetailArray[18],mdDetailArray[19],
//					mdDetailArray[20],mdDetailArray[21],instrumentName,mdDetailArray[22],
//					mdDetailArray[23],mdDetailArray[24],mdDetailArray[25],mdDetailArray[26],mdDetailArray[27],mdDetailArray[28],mdDetailArray[29]
//			);
////			LOGGER.info("收到之前的值："+mdDetailArray[2]+",时间："+mdDetailArray[15]+",最新价："+mdDetailArray[3]);
//			//主力合约列表
//			SocketJsonData<CThostFtdcDepthMarketDataField> socketJsonData=new SocketJsonData<CThostFtdcDepthMarketDataField>(Constant.TICK,marketDataField);
//			//深度行情json
//			Object marketDataFieldToJson= JSONObject.toJSON(socketJsonData);
//
//			//全局推送行情 文本推送
////			ChannelGroups.broadcast(new TextWebSocketFrame(marketDataFieldToJson.toString()));
//			//二进制推送 推送压缩后的包
//			byte [] data= GZIPUtils.compress(marketDataFieldToJson.toString());
//			ByteBuf byteBuf = Unpooled.buffer(data.length * 3);
//			byteBuf.writeBytes(data);
//			ChannelGroups.broadcast(new BinaryWebSocketFrame(byteBuf));
//			LOGGER.info(socketJsonData.getData().getInstrumentID() + ":socket推送的格式：" + marketDataFieldToJson);
////			LOGGER.info("打印推出的数据合约：" + socketJsonData.getData().getInstrumentID() + ",时间：" + socketJsonData.getData().getUpdateTime() + ",最新价：" + socketJsonData.getData().getLastPrice());
//			/**主力合约json**/
//			mdInfoService.sendRedisLastPrice(marketDataField);
////			Object mainContractToJson=JSONObject.toJSON(marketDataField);
//			/**推送完成之后更新首页主力合约列表的合约最新值**/
////			redisTemplate.opsForHash().put(Constant.MAIN_CONTRACT_KEY,marketDataField.getInstrumentID(),mainContractToJson.toString());
//		}
//	}
//
//	private void insertMongoKlineData(String itme,String collectionName){
//		try {
//			//kLineItem|K线类型|code|curTime(yyyy-MM-dd HH:mm:ss)|open|close|high|low|volume|tradeAmt|positionNum|productCode|productMonth
//			System.err.println("KLineItme:"+itme);
//			String [] kLineArray=itme.split("\\|");
//			String code=kLineArray[2],curTime=kLineArray[3],open=kLineArray[4],close=kLineArray[5],high=kLineArray[6],low=kLineArray[7], volume = kLineArray[8],
//					tradeAmt=kLineArray[9], positionNum=kLineArray[10],productCode = kLineArray[11], productMonth=kLineArray[12], avgPrc = kLineArray[13];
//			KLineEntity kLineEntity = new KLineEntity();
//			kLineEntity.setCode(code);
//			kLineEntity.setOpen(Double.valueOf(open));
//			kLineEntity.setClose(Double.valueOf(close));
//			kLineEntity.setHigh(Double.valueOf(high));
//			kLineEntity.setLow(Double.valueOf(low));
//			kLineEntity.setVolume(Integer.valueOf(volume));
//			kLineEntity.setTradeAmt(Double.valueOf(tradeAmt));
//			kLineEntity.setPositionNum(Double.valueOf(positionNum));
//			kLineEntity.setAvgPrc(Double.valueOf(avgPrc));
//			kLineEntity.setCreateTime(DateUtils.parseToDateTimeStr(new Date()));
//			/**mongodb+8小时 时区问题**/
//			Date curTimeDates=DateUtils.addHour(DateUtils.convertStringToDate("yyyy-MM-dd HH:mm:ss", curTime), 8);
//			kLineEntity.setCurTime(curTimeDates);
//			kLineEntity.setProductCode(productCode);
//			kLineEntity.setProductMonth(productMonth);
//			mongoTemplate.insert(kLineEntity,collectionName);
//		}catch (Exception e){
//			e.printStackTrace();
//			LOGGER.info("insertMongoKlineData is error......");
//		}
//	}
//}
